package day10

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * 第2章	DStream入门
 *
 * 2.1	WordCount案例实操
 *
 * 1)	需求：使用netcat工具向9999端口不断的发送数据，通过SparkStreaming读取端口数据并统计不同单词出现的次数
 *
 * 2.2	WordCount解析
 *
 * Discretized Stream是Spark Streaming的基础抽象，代表持续性的数据流和经过各种Spark算子操作后的结果数据流。在内部实现上，DStream是一系列
 * 连续的RDD来表示,每个RDD含有一段时间间隔内的数据，对这些 RDD的转换是由Spark引擎来计算的， DStream的操作隐藏了大多数的细节,
 * 然后给开发者提供了方便使用的高级 API如下图：
 *
 * 例如：水位监控
 *
 * 2.3	几点注意
 *
 * 	一旦StreamingContext已经启动, 则不能再添加新的 streaming computations
 * 	一旦一个StreamingContext已经停止(StreamingContext.stop()), 他也不能再重启
 * 	在一个 JVM 内, 同一时间只能启动一个StreamingContext
 * 	stop() 的方式停止StreamingContext, 也会把SparkContext停掉. 如果仅仅想停止StreamingContext, 则应该这样: stop(false)
 * 	一个SparkContext可以重用去创建多个StreamingContext, 前提是以前的StreamingContext已经停掉,并且SparkContext没有被停掉
 */
object Spark_Streaming_2 {
  def main(args: Array[String]): Unit = {

    // 1、创建配置文件对象 注意：Streaming程序至少不能设置为local，至少需要2个线程
    val conf: SparkConf = new SparkConf().setAppName("SparkStreaming").setMaster("local[*]")

    // 2、创建Spark Streaming上下文环境对象
    val ssc: StreamingContext = new StreamingContext(conf, Seconds(3))

    // 3、操作数据源-从端口中获取一行数据
    val socketDS: ReceiverInputDStream[String] = ssc.socketTextStream(hostname = "hadoop102", port = 9999)

    // 4、对获取的一行数据进行扁平化操作
    val flatMapDS: DStream[String] = socketDS.flatMap(_.split(" "))

    // 5、结构转换
    val mapDS: DStream[(String, Int)] = flatMapDS.map((_, 1))

    // 6、对数据进行聚合
    val reduceDS: DStream[(String, Int)] = mapDS.reduceByKey(_ + _)

    // 7、调用DS的print()，输出结果
    reduceDS.print()

    // 8、启动采集器
    ssc.start()

    // 默认情况下，上下文对象不能关闭
    // ssc.stop()

    // 9、等待采集结束，终止上下文环境对象
    ssc.awaitTermination()
  }
}
